package com.androwit.engine.cassandra;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.Set;

import me.prettyprint.hector.api.Serializer;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

import com.androwit.engine.ItemInfo;
import com.androwit.engine.mapping.Converter;
import com.androwit.engine.mapping.MultipleColumnField;

/**
 * A basic class for transforming a value into multiple columns
 * 
 * @param <F>
 *          the type of a collection element as it should be serialized through {@link Serializer<F>}
 * 
 * 
 * @author hamdi
 */
public abstract class HectorMultiColumnMappingField<J, K, V extends Collection<T>, T, F, I> extends
		HectorMappingField<J, K, V> implements MultipleColumnField<J, K, V, T, I> {

	@SuppressWarnings("unchecked")
	public HectorMultiColumnMappingField(K multiColumnPrefix, Converter<J, V> converter, String name,
			Serializer<K> keySerializer, Serializer<F> valueSerializer) {
		// (Serializer<V>)valueSerializer; is just a way to compile: the
		// converter is actually a (Serializer<T>) used to serialize each element in
		// the list
		super(multiColumnPrefix, converter, name, keySerializer, (Serializer<V>) valueSerializer, ComparatorType.BYTESTYPE,
				false);
	}

	public V parseValues(Map<K, V> values) {
		V toReturn = createVCollection();
		Set<K> keySet = values.keySet();
		for (K key : keySet) {
			if (isMultiColumn(key))
				toReturn.add(fromValue(key, unserializeValue(values, key)));
		}
		return (V) toReturn;
	}

	@SuppressWarnings("unchecked")
	protected F unserializeValue(Map<K, V> values, K key) {
		return (F) valueSerializer.fromByteBuffer(((HColumn<K, V>) values.get(key)).getValueBytes());
	}

	/**
	 * 
	 * creates the <V> Collection
	 * 
	 * @return
	 */
	@SuppressWarnings("unchecked")
	protected V createVCollection() {
		V toReturn = (V) new ArrayList<T>();
		return toReturn;
	}

	@Override
	@SuppressWarnings("unchecked")
	public ItemInfo<Mutator<K>, I> writeValue(ItemInfo<?, I> item, K key, V values, String columnFamily) {
		for (T dynamicColumn : values) {
			HColumn<K, F> column = HFactory.createColumn(toKey(dynamicColumn), toValue(dynamicColumn), keySerializer,
					(Serializer<F>) valueSerializer);
			((ItemInfo<Mutator<K>, I>) item).setOutput(((ItemInfo<Mutator<K>, I>) item).getOutput().addInsertion(key,
					columnFamily, column));
		}
		return (ItemInfo<Mutator<K>, I>) item;
	}

	//	
	// @SuppressWarnings("unchecked")
	// public ItemInfo<Mutator<K>, I> writeValue(ItemInfo<Mutator<K>, I> item,
	// K key, V values, String columnFamily) {
	// for (T dynamicColumn : values) {
	// HColumn<K, F> column = HFactory.createColumn(toKey(dynamicColumn),
	// toValue(dynamicColumn), keySerializer,
	// (Serializer<F>) valueSerializer);
	// item.setOutput(item.getOutput().addInsertion(key, columnFamily, column));
	// }
	// return item;
	// }

	/**
	 * 
	 * constructs a dynamic key or the current column
	 * 
	 * @param dynamicColumn
	 *          column value
	 * @return a key
	 */
	protected abstract K toKey(T dynamicColumn);

	/**
	 * detects if this column is part of a multiple column value
	 * 
	 * @param key
	 * @return true if is multiple column
	 */
	protected abstract boolean isMultiColumn(K key);

	/**
	 * constructs the value content for the column persistent format
	 * 
	 * @param dynamicColumn
	 *          column intermediary list item value
	 * @return the value to write
	 */
	protected abstract F toValue(T dynamicColumn);

	/**
	 * constructs the value content from the column persistent format: ready to be inserted in the value list (before
	 * being passed to the {@link Converter}
	 * 
	 * @param dynColumnValue
	 *          persistent format value
	 * @param value
	 *          the created value
	 * @return the java intermediary list item value
	 */
	protected abstract T fromValue(K key, F dynColumnValue);
}
